// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.utils.hbase;

import com.google.common.base.Joiner;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.stumbleupon.async.Deferred;
import net.opentsdb.utils.Assert;
import net.opentsdb.utils.enums.FamQuls;
import org.hbase.async.*;
import org.hbase.async.Scanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 处理tag树形结构表单元类
 *
 */
public class HbaseTagUtils extends HbaseAbsUtils{

    private static final Logger LOG = LoggerFactory.getLogger(HbaseTagUtils.class);


    /** tsdb meta 计数使用的 tsuid, 之前用000010000100001可能会与单tag时间线重复*/
    private static final String TAG_STATIC_KEY = "00001000010000100001";
    /** The start row to scan on empty search strings.  `!' = first ASCII char. */
    private static final byte[] START_ROW = new byte[] { '!' };

    /** The end row to scan on empty search strings.  `~' = last ASCII char. */
    private static final byte[] END_ROW = new byte[] { '~' };

    public HbaseTagUtils(final HBaseClient client, final byte[] table){
        super(client, table);
    }

    /**
     * 保存树形结构
     *
     * @param metric
     * @param tags
     */
    public void storeTagTree(String metric, Map<String, String> tags){
        String tagk,tagv;

        for(Map.Entry<String,String> entry: tags.entrySet()){

            tagk = entry.getKey();
            tagv = entry.getValue();

            this.storeTagMetricData(metric.getBytes(CHARSET));
            this.storeTagTagkData(metric.getBytes(CHARSET), tagk.getBytes(CHARSET));
            this.storeTagTagvData(metric.concat(tagk).getBytes(CHARSET), tagv.getBytes(CHARSET));
        }
    }


    /**
     * @see #storeTagData(byte[], FamQuls, byte[])
     * */
    public void storeTagMetricData(byte[] qulitifer){
        this.storeTagData(TAG_STATIC_KEY.getBytes(CHARSET), FamQuls.TAG.FAMILY.Mtric, qulitifer);
    }

    /**
     * @see #storeTagData(byte[], FamQuls, byte[])
     * */
    public void storeTagTagkData(byte[] rowkey, byte[] qulitifer){
        this.storeTagData(rowkey, FamQuls.TAG.FAMILY.Tagk, qulitifer);
    }

    /**
     * @see #storeTagData(byte[], FamQuls, byte[])
     * */
    public void storeTagTagvData(byte[] rowkey, byte[] qulitifer){
        this.storeTagData(rowkey, FamQuls.TAG.FAMILY.Tagv, qulitifer);
    }

    /**
     * 保存数据至tsdb-tag表
     *
     * @param rowkey        rowkey
     * @param qulitifer     qulitifer name
     */
    public void storeTagData(byte[] rowkey, FamQuls families, byte[] qulitifer){
        storeData(rowkey,
                families.getByte(),
                qulitifer,
                "".getBytes(CHARSET));
    }

    /**
     * 删除tsuid对应时间线
     *
     * @see #deleteTagTree(String, Map)
     * */
    /*public void deleteTagTree(byte[] tsuid){
        Assert.notNull(tsuid, "tsuid con't be null");

        TsdbPair<String, Map<String,String>> pair = TsuidExchange.getMetricAndTagFromTSUID(tsdb,tsuid);

        this.deleteTagTree(pair.getKey(), pair.getValue());
    }*/

    /**
     * 删除metric所有tag信息
     *
     * @param metric
     */
    public void deleteTagTree(String metric){
        Assert.operation(!Strings.isNullOrEmpty(metric),"metric can't be null");

        List<List<KeyValue>> tagks = scanMetricTag(metric);

        LOG.debug("delete tag metric  {} all tagk/v，tagk size {}",metric,tagks.size());
        if(tagks != null && tagks.size() > 0){
            for (List<KeyValue> list :tagks){
                if(list != null &&list.size() > 0){
                    this.deleteTagTagvData(list.get(0).key(), null);
                }
            }
        }

        this.deleteTagMetricData(metric.getBytes(CHARSET));
        this.deleteTagTagkData(metric.getBytes(CHARSET), null);
    }

    /**
     * 删除tagk的树形结构
     *
     * @param metric
     * @param tagk
     */
    public void deleteTagTree(String metric, String tagk){
        Assert.operation(!Strings.isNullOrEmpty(metric),"metric can't be null");
        Assert.operation(!Strings.isNullOrEmpty(tagk),"tagk can't be null");

        final byte[] metricByte = metric.getBytes(CHARSET);

        DeleteRequest delTagk = new DeleteRequest(table,metricByte, FamQuls.TAG.FAMILY.Tagk.getByte(), tagk.getBytes(CHARSET));
        delTagk.setBufferable(false);
        client.delete(delTagk);

//        LOG.debug(" scan metric and tagk : {}", Joiner.on(",").join(scanMetricTag(metric, FamQuls.TAG.FAMILY.Tagk.getName())));
//        if(!this.existsTagsTagkData(metricByte))
        if(scanMetricTag(metric, FamQuls.TAG.FAMILY.Tagk.getName()).size() == 0 || !this.existsTagsTagkData(metricByte))
            this.deleteTagMetricData(metricByte);

        this.deleteTagTagvData(metric.concat(tagk).getBytes(CHARSET), null);
    }

    /**
     * 删除时间线对应的树形结构
     *
     * @param metric
     * @param tags
     */
    public void deleteTagTree(String metric, Map<String, String> tags){

        Assert.operation(!Strings.isNullOrEmpty(metric),"metric can't be null");
        Assert.operation(tags!=null || tags.size() > 0,"tags can't be null");

        final byte[] metricByte = metric.getBytes(CHARSET);

        for(Map.Entry<String,String> entry: tags.entrySet()){
            this.deleteTagTagvData(metric.concat(entry.getKey()).getBytes(CHARSET),
                    entry.getValue().getBytes(CHARSET));
        }

        for(String tagk: tags.keySet()){
            if(!this.existsTagsTagvData(metric.concat(tagk).getBytes(CHARSET)))
                this.deleteTagTagkData(metricByte, tagk.getBytes(CHARSET));
        }

        if(!this.existsTagsTagkData(metricByte))
            this.deleteTagMetricData(TAG_STATIC_KEY.getBytes(CHARSET));


    }

    /**
     * @see #deleteTagData(byte[], FamQuls, byte[])
     * */
    public void deleteTagMetricData(byte[] qulitifer){
        deleteTagData(TAG_STATIC_KEY.getBytes(CHARSET),
                FamQuls.TAG.FAMILY.Mtric,
                qulitifer);
    }

    /**
     * @see #deleteTagData(byte[], FamQuls, byte[])
     * */
    public void deleteTagTagkData(byte[] rowkey, byte[] qulitifer){
        deleteTagData(rowkey, FamQuls.TAG.FAMILY.Tagk, qulitifer);
    }

    /**
     * @see #deleteTagData(byte[], FamQuls, byte[])
     * */
    public void deleteTagTagvData(byte[] rowkey, byte[] qulitifer){
        deleteTagData(rowkey, FamQuls.TAG.FAMILY.Tagv, qulitifer);
    }


    /**
     * 删除tag表数据
     *
     * @param key
     * @param families
     * @param qulitifer
     */
    public void deleteTagData(byte[] key, FamQuls families, byte[] qulitifer){
        deleteData(key,
                families.getName().getBytes(CHARSET),
                qulitifer);
    }

    /**
     * 判断metric是否还有tagk
     *
     * @see #existsTagsData(byte[], FamQuls, byte[])
     * */
    public boolean existsTagsTagkData(byte[] rowkey){
        return this.existsTagsData(rowkey, FamQuls.TAG.FAMILY.Tagk, null);
    }

    /**
     * 判断tagv下是否还有tagk
     *
     * @see #existsTagsData(byte[], FamQuls, byte[])
     * */
    public boolean existsTagsTagvData(byte[] rowkey){
        return this.existsTagsData(rowkey, FamQuls.TAG.FAMILY.Tagv, null);
    }

    /**
     * 判断tag表数据是否存在
     *
     * @param rowkey
     * @param families
     * @param qulitifer
     * @return
     */
    public boolean existsTagsData(byte[] rowkey, FamQuls families, byte[] qulitifer){
        return existsData(rowkey,families.getByte(), qulitifer);
    }


    /**
     * 获取所有metric
     * @see #fetchTagsQulitifers(byte[], FamQuls)
     * */
    public Set<String> fetchTagsMetrics(){
        byte[] rowkey = TAG_STATIC_KEY.getBytes(CHARSET);
        return fetchTagsQulitifers(rowkey, FamQuls.TAG.FAMILY.Mtric);
    }

    /**
     * 获取metric所有tagk
     * @see #fetchTagsQulitifers(byte[], FamQuls)
     * */
    public Set<String> fetchTagsTagks(String metric){
        byte[] rowkey = metric.getBytes(CHARSET);
        return fetchTagsQulitifers(rowkey, FamQuls.TAG.FAMILY.Tagk);
    }

    /**
     * 获取
     * @see #fetchTagsQulitifers(byte[], FamQuls)
     * */
    public Set<String> fetchTagsTagvs(String metric, String tagk){
        byte[] rowkey = metric.concat(tagk).getBytes(CHARSET);
        return fetchTagsQulitifers(rowkey, FamQuls.TAG.FAMILY.Tagv);
    }

    /**
     * 根据metric查询所有tagk/tagv队列
     *
     * @param metric
     * @return
     */
    public Map<String, Set<String>> fetchTagsTagvs(String metric){
        List<List<KeyValue>> results = this.scanMetricTag(metric);
        Map<String, Set<String>> map = new HashMap<>();
        String tagk;

        if(results != null && results.size() > 0){
            for(List<KeyValue> list: results){
                Set<String> set = new HashSet<>();
                if(list != null && list.size() > 0){

                    tagk = new String(list.get(0).key(), CHARSET);

                    // 截取字符
                    if(tagk.length() > metric.length()){
                        tagk = tagk.substring(metric.length(), tagk.length());
                    }else {
                        //去除tagk信息
                        continue;
                    }

                    list.stream().forEach(key -> set.add(new String(key.qualifier(), CHARSET)));
                    map.put(tagk, set);
                }
            }

        }

        return map;
    }

    /**
     * @see #fetchTagsData(byte[], FamQuls, byte[])
     * */
    public Set<String> fetchTagsQulitifers(byte[] rowkey,FamQuls families){
        Assert.notNull(rowkey,"rowkey con't be null");

        Set<String> tagvs = new HashSet<>();

        List<KeyValue> values = this.fetchTagsData(rowkey,families,null);

        for(KeyValue value: values)
            tagvs.add(new String(value.qualifier(),CHARSET));

        return tagvs;
    }

    /**
     * 获取tree表数据
     *
     * @param rowkey
     * @param families
     * @param qulitifer
     * @return
     */
    public List<KeyValue> fetchTagsData(byte[] rowkey, FamQuls families, byte[] qulitifer){

        return fetchData(rowkey, families.getByte(), qulitifer);
    }

    /**
     * 扫描获取metric所有的tagk、tagv
     *
     * @param rowkey
     * @return
     */
    public List<List<KeyValue>> scanMetricTag(String rowkey){
        return scanMetricTag(rowkey, null);
    }

    /***
     * 扫描指定 column family , cf应支持多个
     *
     * @param rowkey
     * @param familys
     * @return
     */
    public List<List<KeyValue>> scanMetricTag(String rowkey, String familys){
        ArrayList<List<KeyValue>> results = new ArrayList<>();
        ArrayList<ArrayList<KeyValue>> rows;
        byte[] key = rowkey.getBytes(StandardCharsets.UTF_8);
        Scanner scanner = client.newScanner(table);

        try{
            // 使用keyFilter，解决中文扫描问题
            scanner.setFilter(new KeyRegexpFilter(key, StandardCharsets.UTF_8));

            if(familys != null) scanner.setFamilies(familys);

            while ((rows=scanner.nextRows().joinUninterruptibly()) != null){
                results.addAll(rows);
            }
        }catch (Exception e){
            LOG.error("error scan table:",e);
        }finally {
            if(scanner != null)
                scanner.close();
            return results;
        }
    }

    /**
     * 拼接byte[]
     *
     * @param source
     * @param append
     * @return
     */
    private byte[] byteConcat(byte[] source, byte[] append){
        ByteArrayOutputStream output = new ByteArrayOutputStream();

        try{
            output.write(source);
            output.write(append);
        }catch (Exception e){
            LOG.error("concat two byte[] error:",e);
        }

        return output.toByteArray();
    }

}
